package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo12Student {
  def main(args: Array[String]): Unit = {
    /**
     * 统计总分年级排名前十学生各科的分数
     * 1、计算学生的总分
     *
     */

    val conf = new SparkConf()
    conf.setAppName("Demo12Student")
    conf.setMaster("local")

    val sc = new SparkContext(conf)

    //1、读取分数表
    val scoreRDD: RDD[(String, String, Double)] = sc
      .textFile("data/score.txt") //读取文件
      .map(sco => sco.split(",")) //切分数据
      .filter(arr => arr.length == 3) //过滤脏数据
      .map {
        //整理数据，取出字段
        case Array(sid: String, cid: String, sco: String) =>
          (sid, cid, sco.toDouble)
      }


    //2、计算每个学生的总分
    val sumScoRDD: RDD[(String, Double)] = scoreRDD
      .map {
        //取出学号和分数
        case (sid: String, _, sco: Double) =>
          (sid, sco)
      }
      .reduceByKey((x, y) => x + y)


    //3、按照总分排名，取前十的学生
    val top10Array: Array[(String, Double)] = sumScoRDD
      .sortBy(kv => -kv._2)
      .take(10)

    //4、取出前十学生二点学号
    val top10Ids: Array[String] = top10Array.map(kv => kv._1)


    //5、取出前十学生各科的分数
    val top10RDD: RDD[(String, String, Double)] = scoreRDD.filter {
      case (id: String, _, _) =>
        //判断是否在前十学生中
        top10Ids.contains(id)
    }

    top10RDD.foreach(println)

  }

}
